#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <pthread.h>
#include <sys/types.h>
#include <time.h>
#include <unistd.h>

#include <amqp.h>
#include <amqp_tcp_socket.h>

#include <assert.h>

#include "utils.h"

void *work1(void *arg) {
  // amqp_connection_state_t conn = (amqp_connection_state_t)arg;
  char hostname[128] = "localhost";
  int port = 5672, status;
  amqp_socket_t *socket = NULL;
  amqp_connection_state_t conn;

//   if (argc < 4) {
//     fprintf(stderr, "Usage: amqp_listenq host port queuename\n");
//     return 1;
//   }

  char queuename[128] = "slave-health-mem";

  conn = amqp_new_connection();

  socket = amqp_tcp_socket_new(conn);
  if (!socket) {
    die("creating TCP socket");
  }

  status = amqp_socket_open(socket, hostname, port);
  if (status) {
    die("opening TCP socket");
  }

  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                               "admin", "admin"),
                    "Logging in");
  amqp_channel_open(conn, 1);
  die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");


  for (;;) {
    amqp_basic_consume(conn, 1, amqp_cstring_bytes(queuename), amqp_empty_bytes,
                       0, 0, 0, amqp_empty_table);
    die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
    amqp_rpc_reply_t res;
    amqp_envelope_t envelope;

    amqp_maybe_release_buffers(conn);

    struct timeval timeInter;
    timeInter.tv_sec = 0;
    timeInter.tv_usec = 0;

    res = amqp_consume_message(conn, &envelope, NULL, 0);

    if (AMQP_RESPONSE_NORMAL != res.reply_type) {
      break;
    }
    printf("work1:\n");
    printf("Delivery %u, exchange %.*s routingkey %.*s\n",
           (unsigned)envelope.delivery_tag, (int)envelope.exchange.len,
           (char *)envelope.exchange.bytes, (int)envelope.routing_key.len,
           (char *)envelope.routing_key.bytes);

    if (envelope.message.properties._flags & AMQP_BASIC_CONTENT_TYPE_FLAG) {
      printf("Content-type: %.*s\n",
             (int)envelope.message.properties.content_type.len,
             (char *)envelope.message.properties.content_type.bytes);
    }
    printf("----\n");

    amqp_dump(envelope.message.body.bytes, envelope.message.body.len);
    amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
    amqp_destroy_envelope(&envelope);
  }

}

void *work2(void *arg) {
  // amqp_connection_state_t conn = (amqp_connection_state_t)arg;
  char hostname[128] = "localhost";
  int port = 5672, status;
  amqp_socket_t *socket = NULL;
  amqp_connection_state_t conn;

//   if (argc < 4) {
//     fprintf(stderr, "Usage: amqp_listenq host port queuename\n");
//     return 1;
//   }

  char queuename[128] = "slave-health-mem";

  conn = amqp_new_connection();

  socket = amqp_tcp_socket_new(conn);
  if (!socket) {
    die("creating TCP socket");
  }

  status = amqp_socket_open(socket, hostname, port);
  if (status) {
    die("opening TCP socket");
  }

  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                               "admin", "admin"),
                    "Logging in");
  amqp_channel_open(conn, 1);
  die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");


  for (;;) {
    amqp_basic_consume(conn, 1, amqp_cstring_bytes(queuename), amqp_empty_bytes,
                       0, 0, 0, amqp_empty_table);
    die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
    amqp_rpc_reply_t res;
    amqp_envelope_t envelope;

    amqp_maybe_release_buffers(conn);

    struct timeval timeInter;
    timeInter.tv_sec = 0;
    timeInter.tv_usec = 0;

    res = amqp_consume_message(conn, &envelope, NULL, 0);

    if (AMQP_RESPONSE_NORMAL != res.reply_type) {
      break;
    }
    printf("work2:\n");
    printf("Delivery %u, exchange %.*s routingkey %.*s\n",
           (unsigned)envelope.delivery_tag, (int)envelope.exchange.len,
           (char *)envelope.exchange.bytes, (int)envelope.routing_key.len,
           (char *)envelope.routing_key.bytes);

    if (envelope.message.properties._flags & AMQP_BASIC_CONTENT_TYPE_FLAG) {
      printf("Content-type: %.*s\n",
             (int)envelope.message.properties.content_type.len,
             (char *)envelope.message.properties.content_type.bytes);
    }
    printf("----\n");

    amqp_dump(envelope.message.body.bytes, envelope.message.body.len);
    amqp_basic_ack(conn, 1, envelope.delivery_tag, 0);
    amqp_destroy_envelope(&envelope);
  }

}

int main(int argc, char const *const *argv) {
{
 
    pthread_t message_tid;
  pthread_create(&message_tid, NULL, work1, NULL);
}

{
 
  pthread_t message_tid;
  pthread_create(&message_tid, NULL, work2, NULL);
}

  sleep(10000000);

  // die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS),
  //                   "Closing channel");
  // die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS),
  //                   "Closing connection");
  // die_on_error(amqp_destroy_connection(conn), "Ending connection");

  return 0;
}
